#include "lsp_client.h"

#ifndef GLOBAL_EPOCH_VARS
#define GLOBAL_EPOCH_VARS
double epoch_lth = _EPOCH_LTH;
int epoch_cnt = _EPOCH_CNT;
double drop_rate = _DROP_RATE;
#endif

/*
 *
 *
 *				LSP RELATED FUNCTIONS
 *
 *
 */  

 using namespace std;

void lsp_set_epoch_lth(double lth){epoch_lth = lth;}
void lsp_set_epoch_cnt(int cnt){epoch_cnt = cnt;}
void lsp_set_drop_rate(double rate){drop_rate = rate;}


/*
 *
 *
 *				CLIENT RELATED FUNCTIONS
 *
 *
 */  


lsp_client* lsp_client_create(const char* src, int port)
{
	lsp_client *new_lsp = new lsp_client();

	//Initilize data
	new_lsp->sockfd = -1;
	new_lsp->connid = 0;
	new_lsp->seqnum = 0;
	new_lsp->close_flag = false;
	new_lsp->epochs_since_contact = 0;
	new_lsp->waiting_on_ack = true;
	new_lsp->connection_lost = false;
	new_lsp->read_lock = PTHREAD_MUTEX_INITIALIZER;
	new_lsp->to_send_lock = PTHREAD_MUTEX_INITIALIZER;
	new_lsp->new_msg_pushed = PTHREAD_COND_INITIALIZER;
	new_lsp->last_sent = LSPMESSAGE__INIT;

	//Get socket file descriptor and save to fd
	if((new_lsp->sockfd = socket(AF_INET, SOCK_DGRAM, 0)) < 0)
	{
		printf("Failed to get Socket FD while creating lsp client.\n");
		delete new_lsp; new_lsp = NULL;
		return NULL;
	}
	
	//Setup connection info for server
	new_lsp->server.sin_family = AF_INET;
	new_lsp->hp = gethostbyname(src);
	new_lsp->server.sin_port = htons(port);
	if(new_lsp->hp == 0)
	{
		printf("Failed to get host while creating client.\n");
		delete new_lsp; new_lsp = NULL;
		return NULL;
	}
	bcopy((char*) new_lsp->hp->h_addr, (char*) &(new_lsp->server.sin_addr), new_lsp->hp->h_length);
	new_lsp->server.sin_port = htons(port);

	//Set socket to non-blocking mode
	/*if(fcntl(new_lsp->sockfd, F_SETFL, O_NONBLOCK, 1) == -1)
	{
		printf("Failed set socket as non-blocking while creating client.\n");
		delete new_lsp; new_lsp = NULL;
		return NULL;
	}*/

	//Send first message to server:
	new_lsp->last_sent.connid = new_lsp->connid;
	new_lsp->last_sent.seqnum = new_lsp->seqnum;
	new_lsp->last_sent.payload = {(int) 0, NULL};
	lsp_client_send_last(new_lsp);

	//Start lsp_client thread:
	pthread_create(&(new_lsp->lsp_thread), NULL, lsp_client_lsp_thread, (void*) new_lsp);
	pthread_create(&(new_lsp->epoch_thread), NULL, lsp_client_epoch_thread, (void*) new_lsp);

	return new_lsp;
}

int lsp_client_read(lsp_client* a_client, uint8_t* pld)
{
	if(a_client->connection_lost)
	{
		return -1;
	}

	//Wait for a message to be in the queue
	pthread_mutex_lock(&(a_client->read_lock));
	while(a_client->read.size() < 1)
	{
		pthread_cond_wait(&(a_client->new_msg_pushed), &(a_client->read_lock));
	}	
	
	//Get lth and copy data to pld
	int lth = a_client->read.front().payload.len;
	memcpy(pld, a_client->read.front().payload.data, lth);

	//Pop the message and unlock
	delete a_client->read.front().payload.data;
	a_client->read.pop();
	pthread_mutex_unlock(&(a_client->read_lock));

	return lth;
}

bool lsp_client_write(lsp_client* a_client, uint8_t* pld, int lth)
{
	if(a_client->connection_lost)
	{
		return false;
	}

	//Build LSPMessage
	LSPMessage msg = LSPMESSAGE__INIT;
	msg.connid = a_client->connid;
	msg.seqnum = 0;			//This will be changed before sending
	msg.payload.len = lth;
	msg.payload.data = (uint8_t*) malloc(lth);
	memcpy(msg.payload.data, pld, lth);

	//Add LSPMessage to queue to be sent
	pthread_mutex_lock(&(a_client->to_send_lock));
	a_client->to_send.push(msg);
	pthread_mutex_unlock(&(a_client->to_send_lock));
	
	return true;
}

bool lsp_client_close(lsp_client* a_client)
{
	a_client->close_flag = true;
	pthread_join(a_client->lsp_thread, NULL);
	pthread_join(a_client->epoch_thread, NULL);
	close(a_client->sockfd);
	delete a_client; a_client = NULL;
	return true;
}

void *lsp_client_lsp_thread(void *client)
{
	lsp_client *a_client;
	a_client = (lsp_client*) client;
	timeval next;
	timeval now;
	
	//Set time of first epoch
	gettimeofday(&next, NULL);
	next.tv_sec += epoch_lth;

	//Keep doing this loop until parent thread calls lsp_client_close()
	while(!a_client->close_flag)
	{
		//Read from socket:
		uint8_t *buf;
		buf = (uint8_t*) malloc(1024);
		sockaddr from;
		socklen_t from_length = sizeof(from);
		
		//Unlock so epoch can run while waiting
		pthread_mutex_unlock(&(a_client->epoch_lock));

		int byte_count = recvfrom(a_client->sockfd, buf, 1024, 0, (sockaddr *)&from, &from_length);
		
		//Lock so epoch can't run while processing message
		pthread_mutex_lock(&(a_client->epoch_lock));

		if(byte_count > 0)
		{
			LSPMessage* msg;
			msg = lspmessage__unpack(NULL, byte_count, buf);   
	  		if (msg == NULL)
	    	{
	      		printf("Error unpacking incoming message\n");
	    	}
	    	else
	    	{
				int* lth;
				lth = (int*) &(msg->payload);
				if(a_client->connid == 0 && msg->connid != 0 && msg->seqnum == 0)
				{
					//First message received from server
					//Set connid and increment seqnum
					a_client->connid = msg->connid;
					a_client->last_sent.connid = msg->connid;
					a_client->seqnum++;
					a_client->waiting_on_ack = false;
					a_client->epochs_since_contact = 0;
					//cout << "First message recv from server. New connid: " << a_client->connid << endl;
				}
				else if(a_client->connid != msg->connid)
				{
					printf("Recvd message with wrong connid\n");
				}
				else if(msg->seqnum < a_client->seqnum)
				{
					//Ignore message. The seqnum is old. We have already received it once.
					a_client->epochs_since_contact = 0;
					//cout << "Recvd old message from server." << endl;
				}
				else if(*lth <= 0 && msg->seqnum == a_client->seqnum)
				{
					//Deal with ack here
					a_client->seqnum++;
					a_client->waiting_on_ack = false;
					a_client->epochs_since_contact = 0;
					//cout << "Recvd ack from server." << endl;
				}
				else if(*lth > 0 && msg->seqnum >= a_client->seqnum)
				{
					//cout << "Recvd payload from server." << endl;

					//Deal with payload message here
					//Add to read queue and signal other thread
					pthread_mutex_lock(&(a_client->read_lock));
					a_client->read.push(*msg);
					pthread_mutex_unlock(&(a_client->read_lock));
					pthread_cond_signal(&(a_client->new_msg_pushed));

					//Build ack message
					a_client->last_sent.seqnum = msg->seqnum;
					a_client->last_sent.connid = a_client->connid;
					a_client->last_sent.payload.len = 0;
					delete a_client->last_sent.payload.data;
					a_client->last_sent.payload.data = NULL;

					//Send ack message
					lsp_client_send_last(a_client);

					a_client->seqnum++;
					a_client->epochs_since_contact = 0;
				}
			}
			//Clean up memory
			delete msg; msg = NULL;	
		}

		//Clean up memory
		delete buf; buf = NULL; 

		//Check to_send queue and send message if possible
		if((a_client->to_send.size() > 0) && !(a_client->waiting_on_ack))
		{
			//cout << "Msg can be sent." << endl;
			//Lock to_send lock
			pthread_mutex_lock(&(a_client->to_send_lock));
			
			//Setup pointers to the message to be sent and last_sent
			LSPMessage* front;
			LSPMessage* last_sent;
			front = &(a_client->to_send.front());
			last_sent = &(a_client->last_sent);

			//Set seqnum and connid
			last_sent->seqnum = a_client->seqnum;
			last_sent->connid = a_client->connid;
			
			//Set payload length
			last_sent->payload.len = front->payload.len;

			//Delete old payload data and copy new stuff over
			delete last_sent->payload.data;
			last_sent->payload.data = (uint8_t*) malloc(front->payload.len);
			memcpy(last_sent->payload.data, front->payload.data, front->payload.len);

			//Delete data before poping from queue
			delete front->payload.data;
			a_client->to_send.pop();

			//Unlock to_send lock
			pthread_mutex_unlock(&(a_client->to_send_lock));

			a_client->waiting_on_ack = true;
			lsp_client_send_last(a_client);
		}
	}
	return NULL;
}

void *lsp_client_epoch_thread(void *client)
{
	lsp_client *a_client;
	a_client = (lsp_client*) client;

	timeval last;
	timeval now;
	double time_left = epoch_lth;
	
	while(!a_client->close_flag)
	{
		//Loop and sleep until the epoch has passed
		gettimeofday(&last, NULL);
		time_left = epoch_lth;
		while(time_left > 0)
		{
			usleep((int) (time_left * 1000000));
			gettimeofday(&now, NULL);
			time_left = epoch_lth - 
				((double) now.tv_sec + ((double) now.tv_usec/1000000)
					- ((double) last.tv_sec + ((double) last.tv_usec/1000000)));
		}

		//Time to run epoch stuff. Get lock.
		pthread_mutex_lock(&(a_client->epoch_lock));

		//epoch has passed
			a_client->epochs_since_contact++;

			if(a_client->epochs_since_contact > epoch_cnt)
			{
				a_client->connection_lost = true;
				a_client->close_flag = true;
			}

			//Send last sent message
			lsp_client_send_last(a_client);

		//Done with epoch. Release epoch lock.
		pthread_mutex_unlock(&(a_client->epoch_lock));
	}
	return NULL;
}

void lsp_client_send_last(lsp_client* a_client)
{
	//Gen random number [0,1] for packet dropping
	double x = ((double) rand()/ (RAND_MAX));
	if(x > drop_rate)
	{
		void* buf;
		size_t len = lspmessage__get_packed_size(&a_client->last_sent);
		buf = malloc(len);
		lspmessage__pack(&a_client->last_sent, (uint8_t*) buf);
		int server_size = sizeof(a_client->server);
		sendto(a_client->sockfd, buf, len, 0, (sockaddr*) &(a_client->server), server_size);

		/*cout << "Sending last now. Seqnum: " << a_client->last_sent.seqnum 
				<< "  connid: " << a_client->last_sent.connid
				<< "  len: " << a_client->last_sent.payload.len << endl;*/

		delete buf;
	}
}
